package com.candy.candycommon.manager;

import com.google.common.collect.Lists;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * @author suYanLong
 * @version 1.0
 * @date 2023/4/25 11:50
 * @Description TODO LockSupport.park(); LockSupport.unpark(); 控制多线程事务提交
 */
@Slf4j
@AllArgsConstructor
public class MultiThreadTransaction {

    @FunctionalInterface
    public interface ConvertCallBack<T> {

        void callBack(List<T> list);

    }

    ThreadPoolTaskExecutor ioDense;

    PlatformTransactionManager transactionManager;

    //-----------多线程事务新增操作-----------
    public <T> void batchSchedule(List<T> addList, ConvertCallBack convertCallBack) {
        if (!CollectionUtils.isEmpty(addList)) {
            //定义局部变量，是否成功、顺序标识、等待线程队列
            AtomicBoolean isSuccess = new AtomicBoolean(true);
            AtomicInteger cur = new AtomicInteger(1);
            List<Thread> unfinishedList = new ArrayList<>();
            //切分新增集合
            List<List<T>> partition = Lists.partition(addList, 1);
            int totalSize = partition.size();
            //多线程处理开始
            CompletableFuture<Void> future =
                    CompletableFuture.allOf(partition.stream().map(addPartitionList -> CompletableFuture.runAsync(() -> {
                        //Spring事务内部由ThreadLocal存储事务绑定信息，因此需要每个线程新开一个事务
                        DefaultTransactionDefinition defGo = new DefaultTransactionDefinition();
                        defGo.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
                        TransactionStatus statusGo = transactionManager.getTransaction(defGo);
                        int curInt = cur.getAndIncrement();
                        try {
                            log.info("当前是第{}个线程开始启动，线程名={}", curInt, Thread.currentThread().getName());
                            convertCallBack.callBack(addPartitionList);
                            log.info("当前是第{}个线程完成批量插入，开始加入等待队列，线程名={}", curInt, Thread.currentThread().getName());
                            //ArrayList线程不安全，多线程会出现数据覆盖，体现为数据丢失
                            synchronized (unfinishedList) {
                                unfinishedList.add(Thread.currentThread());
                            }
                            log.info("当前是第{}个线程已加入队列，开始休眠，线程名={}", curInt, Thread.currentThread().getName());
                            notifyAllThread(unfinishedList, totalSize, false);
                            LockSupport.park();
                            if (isSuccess.get()) {
                                log.info("当前是第{}个线程提交，线程名={}", curInt, Thread.currentThread().getName());
                                transactionManager.commit(statusGo);
                            } else {
                                log.info("当前是第{}个线程回滚，线程名={}", curInt, Thread.currentThread().getName());
                                transactionManager.rollback(statusGo);
                            }
                        } catch (Exception e) {
                            log.error("当前是第{}个线程出现异常，线程名={}", curInt, Thread.currentThread().getName(), e);
                            transactionManager.rollback(statusGo);
                            isSuccess.set(false);
                            notifyAllThread(unfinishedList, totalSize, true);
                        }
                    }, ioDense)).toArray(CompletableFuture[]::new));
            future.join();
        }
    }

    private static void notifyAllThread(List<Thread> unfinishedList, int totalSize, boolean isForce) {
        if (isForce || unfinishedList.size() >= totalSize) {
            log.info("唤醒当前所有休眠线程，线程数={}，总线程数={},是否强制={}", unfinishedList.size(), totalSize, isForce);
            for (Thread thread : unfinishedList) {
                log.info("当前线程={}被唤醒", thread.getName());
                LockSupport.unpark(thread);
            }
        }
    }

}
